package main.Concurrent.Iteration;

import java.util.Iterator;
import java.util.concurrent.CountDownLatch;

import main.Concurrent.Iteration.IterationOperation;
import main.initiate.Configuration;

import com.googlecode.actorom.Actor;
import com.googlecode.actorom.Address;
import com.googlecode.actorom.ExitActorMessage;
import com.googlecode.actorom.ExitReason;
import com.googlecode.actorom.KillActorException;
import com.googlecode.actorom.Topology;
import com.googlecode.actorom.annotation.AddressInstance;
import com.googlecode.actorom.annotation.OnMessage;
import com.googlecode.actorom.annotation.TopologyInstance;

/**
 * Any actor associated with this handler should not be run on a foreign machine
 * so that the latch will work. Thus, there is no need for a constuction message
 * since this can't be run foreign.
 * 
 * @author Jared Hance
 *
 * @param <E>
 * @param <R>
 * @param <A>
 */
public class IterationHandler<E, R, A> {
    @TopologyInstance
    private Topology topology;
    @AddressInstance
    private Address address;
    private Iterable<E> source;
    
    private IterationOperation<E, R, A> operation;
    private CountDownLatch countDownLatch;
    
    private int length;
    
    private A aggregate;
    
    /**
     * Calculates length manually by O(n) transversal
     * @param source
     * @param op
     */
    public IterationHandler(Iterable<E> source, IterationOperation<E, R, A> op) {
        int l = 0;
        for(E e : source) {
            l++;
        }
        
        this.source = source;
        this.operation = op;
        this.countDownLatch = new CountDownLatch(l);
    }
    
    public IterationHandler(Iterable<E> source, IterationOperation<E, R, A> op, int length) {
        this.source = source;
        this.operation = op;
        this.countDownLatch = new CountDownLatch(length);
        this.length = length;
    }
    
    @OnMessage(type = IterationStartMessage.class)
    public void onStart(IterationStartMessage<A> m) {
        this.aggregate = m.getOriginalAggregate();
        int cores = Configuration.getSingleton().getNumberCores();
        int jobsPerCore = length / cores;
        int diff = length % cores;
        
        int index = 0;
        Iterator<E> iter = source.iterator();
        for(int i = 0; i < cores; i++) { // Cores
            int count = jobsPerCore;
            if(i < diff) {
                count++;
            }
            
            Address child = 
                    topology.spawnActor(address.getId() + " Child: " + i,
                            new IterationOperationHandler<E, R, A>());
            Actor childActor = topology.getActor(child);
            childActor.send(new IterationOperationConstructionMessage<E,R,A>(address, operation));
            
            for(int j = 0; j < count; j++) {
                E value = iter.next();
                childActor.send(new IterationOperationRunMessage<E>(index, value));
                index++;
            }
            
            childActor.send(new ExitActorMessage(this.address, ExitReason.KILL));
        }
    }
    
    @OnMessage(type = IterationOperationDoneMessage.class)
    public void onChildFinished(IterationOperationDoneMessage<R> m) {
        R result = m.getResult();
        int index = m.getIndex();
        aggregate = operation.aggregate(aggregate, index, result);
        countDownLatch.countDown();
        System.out.println(countDownLatch.getCount());
        if(countDownLatch.getCount() == 0) {
            KillActorException.throwException();
        }
    }
    
    /** 
     * This can be used to latch onto the operation to finish it from a context
     * not within an actor (so the IterationCompleteMessage cannot be finished)
     * @return The completed operation
     * @throws InterruptedException 
     */
    public A latch() throws InterruptedException {
        this.countDownLatch.await();
        return this.aggregate;
    }
}
